package com.shujia.flink

import org.apache.flink.cep.scala.pattern.Pattern
import org.apache.flink.cep.scala.CEP
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.api.scala._

object Demo5CEP {
  def main(args: Array[String]): Unit = {

    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    /**
      * 1,192.168.0.1,fail
      * 1,192.168.0.2,fail
      * 1,192.168.0.3,fail
      * 1,192.168.0.4,fail
      */
    val loginEventStream = env.socketTextStream("node1", 7777)
      .map(line => {
        val split = line.split(",")
        LoginEvent(split(0), split(1), split(2))
      })

    loginEventStream.print()

    /**
      * 1,192.168.0.1,a
      * 1,192.168.0.2,b
      * 1,192.168.0.2,v
      * 1,192.168.0.3,b
      * 1,192.168.0.4,c
      *
      */

    //定义模板
    val pattern = Pattern

      .begin[LoginEvent]("begin") //匹配第一个事件
      .oneOrMore
      .where(event => event.t.equals("a"))

      .next("next") //匹配下一个事件
      .where(event => event.t.equals("b"))
      .oneOrMore //循环匹配，匹配多个，默认宽松的连续性
      .consecutive() //严格的连续性

      .followedBy("end") //匹配下一个事件
      .where(event => event.t.equals("c"))
      .within(Time.seconds(10)) //如果19秒内没有完成丢弃匹配


    val patternStream = CEP.pattern(loginEventStream.keyBy(_.id), pattern)


    patternStream.select(map => {
      val first = map.getOrElse("begin", null).iterator.toList.toString()
      val second = map.getOrElse("next", null).iterator.toList.toString()
      val end = map.getOrElse("end", null).iterator.toList.toString()

      (first, second, end)
    }).print()

    env.execute()


  }

  case class LoginEvent(id: String, ip: String, t: String)

}
